import com.zhenjun.domin.Person;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;

/**
 * @author wangzj
 * @description 基于数据来源和数据库连接的练习
 * @date 2020/8/1 23:23
 */
public class SourceAndConnectorDemo {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //1、读取本地文件
        //如果是hdfs文件，只需要传hdf路径：hdfs://nnHost:nnPort/path/to/my/textfile
        DataSet<String> data = env.readTextFile("D:\\MyOperation\\flinkdemo\\testfile\\test");

        //2、读取csv文件,逗号分隔数据
        DataSet<Tuple3<Integer, String, Double>> data1 = env.readCsvFile("D:\\MyOperation\\flinkdemo\\testfile\\csvfile.csv")
                .types(Integer.class, String.class, Double.class);
        //2.1、输出指定的2列数据（第二和第三）
        DataSet<Tuple2<String, Double>> data11 = env.readCsvFile("D:\\MyOperation\\flinkdemo\\testfile\\csvfile.csv")
                //0表示不输出，1表示输出，includeFields包含的需要占所有的字段
                .includeFields("011")
                //要输出的类型
                .types(String.class, Double.class);
        //2.2、支持输出类型的指定
        DataSet<Person> data12 = env.readCsvFile("D:\\MyOperation\\flinkdemo\\testfile\\csvfile.csv")
                .pojoType(Person.class, "id", "name", "age");

        //过滤有flink的行
//        data.filter(value -> !value.equals("flink")).printToErr();
        //输出文件内容
        data12.printToErr();

    }

}


